package com.tgy.kafka.Test;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/***
 * @ClassName: MyProducer
 * @Description:
 * @Auther: 送你一块二向箔
 * @Date: 10:04 2022/4/2
 * @version : V1.0
 */
@Slf4j
public class MyProducer {
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,10.31.167.10:9093,10.31.167.10:9094");
//把发送的key从字符串序列化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//把发送消息value从字符串序列化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        //key,作用是决定了往那个topic发送，value具体发送的消息内容  可以指定分区 //未指定发送分区，具体发送的分区计算公式：hash(key)%partitionNum
        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("yaml", "key", "msg");

        producer.send(producerRecord, new Callback() {
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    System.err.println("异步发送消息失败：" +
                            exception.getStackTrace());
                }
                if (metadata != null) {
                    System.out.println("异步方式发送消息结果：" + "topic-" +metadata.topic() + "|partition-"+ metadata.partition() + "|offset-" + metadata.offset());
                }
            }
        });

        Thread.sleep(1000L);
    }

    //同步发送
    public static void syncSend() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
//把发送的key从字符串序列化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//把发送消息value从字符串序列化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        //key,作用是决定了往那个topic发送，value具体发送的消息内容  可以指定分区 //未指定发送分区，具体发送的分区计算公式：hash(key)%partitionNum
        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("yaml",1, "key","msgValue");
        try {
            RecordMetadata metadata = null;
            metadata = producer.send(producerRecord).get();
            log.info("同步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-" + metadata.partition() + "|offset-" + metadata.offset());
        } catch (InterruptedException | ExecutionException e) {
            log.error("发送失败");
            e.printStackTrace();
        }
    }
}


